package com.bml.architect.order;

import com.bml.architect.utils.MqConstants;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * 顺序消息的发送
 * 首先在发送者端：在默认情况下，消息发送者会采取Round Robin轮询方式把消息发送到不同的
 * MessageQueue(分区队列)，而消费者消费的时候也从多个MessageQueue上拉取消息，这种情
 * 况下消息是不能保证顺序的。而只有当一组有序的消息发送到同一个MessageQueue上时，才能
 * 利用MessageQueue先进先出的特性保证这一组消息有序。
 * 而Broker中一个队列内的消息是可以保证有序的。
 *
 * rocketmq的顺序消息只能保证局部有序，没有办法保证全局有序
 * 局部有序指的是范围内的消息发送到同一个队列中保证了有序，范围内的消息发送到了不同的队列中
 * 那么消费的时候是没有办法保证有序的
 */
public class Producer {

    public static void main(String[] args) {
        DefaultMQProducer producer = new DefaultMQProducer("orderMsgGroup");
        try {
            producer.setNamesrvAddr(MqConstants.getNameServerAddr());
            producer.start();
            for (int i = 1; i <= 10; i++) {
                final int orderId = i;
                for (int j = 0; j < 5; j++) {
                    Message msg = new Message("OrderTopicTest", "orderId" + orderId, "key_" + orderId, ("order_" + orderId + " step " + j).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                            //保证在当前循环，也就是第二层循环每次发送到同一个队列MessageQueue中
                            Integer id = (Integer) o;
                            int index = id % list.size();
                            return list.get(index);
                        }
                    }, orderId);
                    System.out.printf("%s%n", sendResult);
                }
            }
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } finally {
            producer.shutdown();
        }

    }
}
